#pragma once
#include "../common/mq_msg/mq_msg.pb.h"
#include "../common/helper.hpp"
#include "../common/log.hpp"
#include "mq_consumer.hpp"
#include "mq_host.hpp"
#include "mq_route.hpp"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../common/mq_proto/mq_proto.pb.h"
#include "../common/mq_threadpool.hpp"

#include <iostream>
#include <functional>
#include <mutex>
#include <memory>

namespace xzt
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

    using openChannelRequestPtr = std::shared_ptr<xzt::openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<xzt::closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<xzt::declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<xzt::deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<xzt::declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<xzt::deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<xzt::queueBindRequest>;
    using queueUnbindRequestPtr = std::shared_ptr<xzt::queueUnbindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<xzt::basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<xzt::basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<xzt::basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<xzt::basicCancelRequest>;


    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
    
        Channel(const std::string& id, 
                const VirtualHost::ptr& host, 
                const ConsumerManager::ptr& p_cm, 
                const muduo::net::TcpConnectionPtr& conn, 
                const ProtobufCodecPtr& codec,
                const threadpool::ptr& threadpool):
            _id(id),_consumer(nullptr),_conn(conn),_codec(codec),
            _p_cm(p_cm),_host(host),_threadpool(threadpool)
        {
            DLOG("channel is created, id: %s", id.c_str());
        }

        ~Channel()
        {
            if(_consumer != nullptr)
            {
                _p_cm->remove(_consumer->_tag, _consumer->_queue_name);
                _consumer = nullptr;
            }
            DLOG("channel is destroyed, id: %s", _id.c_str());
        }
    public:
        // 交换机相关
        void declareExchange(const declareExchangeRequestPtr& req)
        {
            bool ret = _host->declareExchange(req->exchange_name(), req->exchange_type(), req->durable(), req->auto_delete(), req->args());
            if(ret)
            {
                ILOG("%s 交换机创建成功", req->exchange_name().c_str());
            }
            _basicResponse(ret, req->rid(), req->cid());
        }

        void deleteExchange(const deleteExchangeRequestPtr& req)
        {
            _host->deleteExchange(req->exchange_name());
            _basicResponse(true, req->rid(), req->cid());
        }

        // 消息队列相关
        void declareMsgQueue(const declareQueueRequestPtr& req)
        {
            bool ret = _host->declareMsgQueue(req->queue_name(), req->durable(), req->exclusive(), req->auto_delete(), req->args());
            if(ret == false)
            {
                _basicResponse(false, req->rid(), req->cid());
                return;
            }
            // 初始化消费者的队列管理句柄
            _p_cm->declareQueueConsumer(req->queue_name());
            _basicResponse(true, req->rid(), req->cid());
        }

        void deleteMsgQueue(const deleteQueueRequestPtr& req)
        {
            _p_cm->removeQueueConsumer(req->queue_name());
            _host->deleteMsgQueue(req->queue_name());
            _basicResponse(true, req->rid(), req->cid());
        }

        // 绑定相关
        void queueBind(const queueBindRequestPtr& req)
        {
            bool ret = _host->bind(req->exchange_name(), req->queue_name(), req->binding_key());
            _basicResponse(ret, req->rid(), req->cid());
        }

        void queueUnbind(const queueUnbindRequestPtr& req)
        {
            bool ret = _host->unBind(req->exchange_name(), req->queue_name());
            _basicResponse(ret, req->rid(), req->cid());
        }

        // 消息发布 -- {生产者}
        bool basicPublish(const basicPublishRequestPtr& req)
        {
            // 1.判断交换机是否存在
            Exchange::ptr exchange = _host->selectExchange(req->exchange_name());
            if(exchange == nullptr)
            {
                _basicResponse(false, req->rid(), req->cid());
                return false;
            }
            // 2.进行交换路由，判断消息到达哪些交换机绑定的队列
            MsgQueueBindingMap binding_map = _host->getExchangeBindings(req->exchange_name());
            string routing_key;
            BasicPropeties* properties = nullptr;
            if(req->has_properties())
            {
                routing_key = req->properties().routing_key();
                properties = req->mutable_properties();
            }
            for(auto& iter : binding_map)
            {
                DLOG("Channel::basicPublish -> 交换机类型：%d", exchange->_type);
                if(Router::route(exchange->_type, routing_key, iter.second->_binding_key))  // 交换机路由匹配
                {
                    // 3.将消息投递到队列
                    _host->basicPublish(iter.first, properties, req->body());
                    // 4.向线程池中添加消息消费任务 (向指定队列的订阅者推送消息)
                    auto task = std::bind(&Channel::_consume, this, iter.first);
                    _threadpool->push(task);
                }
            }
            _basicResponse(true, req->rid(), req->cid());
            return true;
        }

        // 消息确认
        void basicAck(const basicAckRequestPtr& req)
        {
            _host->basicAck(req->queue_name(), req->message_id());
            _basicResponse(true, req->rid(), req->cid());
        }

        // 订阅消息队列 -- {消费者}
        void basicConsume(const basicConsumeRequestPtr& req)
        {
            // 1.判断队列是否存在
            bool ret = _host->existsMsgQueue(req->queue_name());
            if(ret == false)
            {
                _basicResponse(false, req->rid(), req->cid());
                return;
            }
            // 2.创建队列所属消费者 -- (此时当前Channel角色就是消费者)
            auto cb = std::bind(&Channel::consumerCallback, this, 
                               std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            _consumer = _p_cm->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
            
            _basicResponse(true, req->rid(), req->cid());
        }
        
        // 取消订阅消息队列
        void cancelConsume(const basicCancelRequestPtr& req)
        {
            if(_consumer != nullptr)
            {
                _p_cm->remove(req->consumer_tag(), req->queue_name());
                _consumer = nullptr;
            }
            _basicResponse(true, req->rid(), req->cid());
        }
    private:
        void _basicResponse(bool is_success, const std::string& rid, const std::string& cid)
        {
            xzt::basicCommonReponse response;
            response.set_success(is_success);
            response.set_rid(rid);
            response.set_cid(cid);
            _codec->send(_conn, response);
        }

        // 从指定队列中消费消息
        void _consume(const std::string& queue_name)
        {
            // 1.从队列中获取消息
            MessagePtr msg = _host->basicConsume(queue_name);
            if(msg == nullptr)
            {
                ELOG("Channel::consume -> 队列%s中没有消息", queue_name.c_str());
                return;
            }
            // 2.从消费者队列中获取订阅者
            Consumer::ptr consumer = _p_cm->choose(queue_name);
            if(consumer == nullptr)
            {
                ELOG("Channel::consume -> 消费者订阅队列%s中消费者为空", queue_name.c_str());
                return;
            }
            // 3.调用消费者对应的回调函数，处理消息
            ILOG("Channel::consume -> 消费者%s从队列%s中获取消息, 内容: %s", consumer->_tag.c_str(), queue_name.c_str(), msg->payload().body().c_str());
            consumer->_callback(consumer->_tag, msg->mutable_payload()->mutable_propeties(), msg->payload().body());
            // 4.判断是否需要自动确认
            if(consumer->_auto_ack)
            {
                _host->basicAck(queue_name, msg->payload().propeties().id());
            }
        }

        void consumerCallback(const std::string& tag, const xzt::BasicPropeties* properties, const std::string& body)
        {
            // 将参数组织成推送消费请求，并推送给Channel对应的客户端
            basicConsumeResponse resp;
            resp.set_cid(this->_id);
            resp.set_consumer_tag(tag);
            resp.set_body(body);
            
            if(properties != nullptr)
            {
                resp.mutable_properties()->set_id(properties->id());
                resp.mutable_properties()->set_delivery_mode(properties->delivery_mode());
                resp.mutable_properties()->set_routing_key(properties->routing_key());
            }
            _codec->send(_conn, resp);
        }
    private:
        std::string _id;  // 通道ID
        Consumer::ptr _consumer;  // 关联的消费者
        muduo::net::TcpConnectionPtr _conn;  // 关联的连接
        ProtobufCodecPtr _codec;  // 编解码器
        ConsumerManager::ptr _p_cm;  // 消费者管理器
        VirtualHost::ptr _host;  // 虚拟主机
        threadpool::ptr _threadpool;  // 线程池
    };

    // 通道管理器
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;

        ChannelManager() = default;
        ~ChannelManager() = default;

        bool openChannel(const std::string& id, 
                const VirtualHost::ptr& host, 
                const ConsumerManager::ptr& p_cm, 
                const muduo::net::TcpConnectionPtr& conn, 
                const ProtobufCodecPtr& codec,
                const threadpool::ptr& threadpool)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            if(_channels.find(id) != _channels.end())
            {
                ELOG("ChannelManager::openChannel -> 通道%s已存在", id.c_str());
                return false;
            }

            // 创建通道
            _channels[id] = std::make_shared<Channel>(id, host, p_cm, conn, codec, threadpool);
            return true;
        }

        void closeChannel(const std::string& id)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            _channels.erase(id);
        }

        Channel::ptr getChannel(const std::string& id)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            if(_channels.find(id) == _channels.end())
            {
                return nullptr;
            }
            return _channels[id];
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;  // 通道ID -> 通道映射
    };

}